<?php

namespace app\command;

use app\api\controller\wap\Trade;
use app\command\extend\http\ApiCache;
use app\common\model\coin\CallApi;
use app\common\model\coin\CoinRule;
use app\common\model\coin\CoinRuleRun;
use mall\library\Huobi;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\Exception;
use think\facade\Db;


/**
 * 监控账户详情
 * Class HistoryTrade
 * @package app\command
 */
class MonitorAccount extends Command
{
    /**
     * 初始化的内存表
     */
    const TABLE_LIST = [
        'CONFIG'   => ['config',true,false,1,[      //【表名,是否初始化创建,是否启用锁,记录数（1024的倍数）,字段参数】
            ['value','TYPE_STRING',20],  // 配置的值
        ]], //配置信息
        'USER_DETAILS' => ['user_details',true,false,10,[  //键为用户ID
            ['process_index_name','TYPE_STRING',20],  // 进程索引
            ['account_id','TYPE_STRING',15],  // 第三方api账号
            ['access_key','TYPE_STRING',50],
            ['secret_key','TYPE_STRING',50],
            ['last_time','TYPE_STRING',15], // 最后更新时间
        ]], // 监控用户的密钥详细及进程信息

        'ACCOUNTS_DETAILS'  => ['account_details',true,false,10,[  //键为用户ID
            ['details','TYPE_STRING',20000],  // 详情信息
            ['validate','TYPE_STRING',40],  // 验证字段
            ['amount_query_err_time','TYPE_STRING',15], //失败的时间
            ['last_time','TYPE_STRING',15], // 最后修改时间
        ]]  // 第三返回的账号详情信息
    ];
    const SCALE = 30;   //价格保留的小数点位数

    // 动态的表列表
    static $tableDynamicList = [];

    /**
     * 配置表的键列表
     */
    const CONFIG_TABLE_KEY_LIST = [
    //    'TRADE_REFRESH_CONFIG_KEY'  => 'trade_refresh',  // 判断是否刷新订单状态的键
    ];


    /**
     * 进程参数设置
     */
    const PROCESS_CONFIG = [
        'MONITOR_NUM' => 5,  // 监控进程数（查询账户详情数）
        'SQL_ONCE_MAX_RECORD_NUM'   => 200, //查询mysql时，单次最多获取的记录数
        'MAX_REQUEST_USER'   => 10000,    //设置进程最大请求次数，超出次数则重其进程 【获取用户信息】
        'MAX_REQUEST_MONITOR'   => 10000,    //设置进程最大请求次数，超出次数则重其进程【监控账户详情】
        'TABLE_DYNAMIC_NAME'  => 'USER_DETAILS_',  //动态表名

        'LOG'   => [
            'TYPE_NAME'  => 'MONITORY_ACCOUNT', // 日志类里对应的类型
            'TYPE_CUSTOM_LIST'  => [
                'ADD'   => 'add',   //添加监控
                'DEL'   => 'del',   //删除监控,
                'INIT'  => 'init',   //监控初始化
                'CHANGE'=> 'change',    //数据变更
            ],  // 自定义类型
        ],

    ];

    /**
     * ENV 配置参数KEY列表
     */
    const ENV_KEY_LIST = [
    ];

    /**
     * 间隔时间执行，单位秒
     */
    const INTERVAL_TIME_LIST = [
        'USER_DETAILS_REFRESH'  => 10, // 用户监控刷新时间
    ];

    /**
     * 休眠等待时间，单位微妙
     */
    const USLEEP_TIME_LIST = [
        'POLL_TIME'  => 1000000, //查询一轮之后的间隔时间
        'ERROR_RESTART' => 500000,  //进程失败重启时间

    ];

    /**
     * 通知配置
     * @var array
     */
    static $noticeConfig = [
        'CMD'   => 'balanceChange', // 余额变化
        'IP_LIST'    => '127.0.0.1', // 请求地址
        'TIMEOUT'   => 2,   //设置命令有效时间
    ];


    /**
     * 重写configure
     * {@inheritdoc}
     */
    protected function configure()
    {
        // 命令的名字（"think" 后面的部分）
        $this->setName('MonitorAccount')
            // 配置一个参数 使用$input->getArgument('username')获取
            // ->addArgument('username')
            // 运行 "php think list" 时的简短描述
            ->setDescription('监控账户余额变化情况')
            // 运行命令时使用 "--help" 选项时的完整命令描述
            ->setHelp("一直执行 无参数");
    }


    /**
     *  * 重写execute
     *  * {@inheritdoc}
     *
     * @param Input $input
     * @param Output $output
     */
    public function execute(Input $input, Output $output)
    {

        // 设置 udp 端口
        self::$noticeConfig['IP_LIST'] .= ':' . UdpServer::serverPort();

        /******* 设置数量变化的表 **************/
        for($i = 0;$i < self::PROCESS_CONFIG['MONITOR_NUM'];++$i) {
            $tableName = self::PROCESS_CONFIG['TABLE_DYNAMIC_NAME'] . $i;
            self::$tableDynamicList[$tableName] = self::TABLE_LIST['USER_DETAILS'];
            self::$tableDynamicList[$tableName][0] = strtolower($tableName);
        }
        /***** END 设置数量变化的表 ************/


        // 创建内存表
        static::swooleTable();


        //  开启监控配置信息
        $this->monitorUserList();


        // 监控并比较余额
        $this->compareBalance();


        //开启等待
        Swoole::wait();

    }



    /**
     * 开启进程，间隔获取需要监控余额的用户
     */
    private function monitorUserList() {
        // 开启计算刷新进程，并失败重启
        Swoole::startProcessAndErrorRestart(function(){
            // 刷新需监控的用户信息
            $this->userDetailsRefresh();
            sleep(self::INTERVAL_TIME_LIST['USER_DETAILS_REFRESH']); // 刷新间隔
        },false,self::USLEEP_TIME_LIST['ERROR_RESTART'],self::PROCESS_CONFIG['MAX_REQUEST_USER']);
    }

    /**
     * 监控并对余额进行比较
     */
    private function compareBalance() {
        for($i = 0;$i < self::PROCESS_CONFIG['MONITOR_NUM'];++$i) {
            $tableName = self::PROCESS_CONFIG['TABLE_DYNAMIC_NAME'] . $i;
            Swoole::startProcessAndErrorRestart(function() use($tableName){
                // 获取监控数据中的用户余额，并进行匹配处理
                $table = self::swooleTable($tableName);
                foreach($table as $userId=>$details) {
                    // 查询并监控数据
                    $this->compareBalanceOne($userId,$details);
                }
                usleep(self::USLEEP_TIME_LIST['POLL_TIME']); //  监控刷新频率
            },false,self::USLEEP_TIME_LIST['ERROR_RESTART'],self::PROCESS_CONFIG['MAX_REQUEST_MONITOR']);
        }
    }


    /**
     * 将余额推送到指定位置
     * @param $userId
     * @param $accountId
     * @param $balance
     */
    private function pushBalance($userId,$accountId,$balance) {
        UdpServer::request(UdpServer::CMD_LIST['API_CACHE_CALL'],null,['params'=>ApiCache::encode([
            'type'  => 'setBalance',
            'userId'    => $userId,
            'accountId' => $accountId,
            'balance'   => $balance,
        ])]);
    }


    /**
     * 从第三方获取用户余额信息，并进行匹配处理
     * @param $userId   // 用户ID
     * @param $details  // 用户密钥详情
     */
    private function compareBalanceOne(int $userId,array $details) {
        // 获取用户 余额详情
        if(!$balanceList = (new Huobi($details['account_id'],$details['access_key'],$details['secret_key']))->get_balance($details['account_id'])) {
            return false;
        }
        $accouontTable = self::swooleTable('ACCOUNTS_DETAILS');
        $accountDetails = null;
        $changeData = [];
        if($balanceList['status'] == 'ok') {
            $list = [];
            foreach ($balanceList['data']['list'] as $value) {
                $currency = $value['currency'];
                if(!array_key_exists($currency,$list)) {
                    $list[$currency] = [
                        'currency'  => $currency,
                        'trade'     => 0,
                        'frozen'    => 0,
                    ];
                }
                switch ($value['type']) {
                    case 'frozen':
                    case 'trade':
                        $list[$currency][$value['type']] = $value['balance'];
                        break;
                }
            }
            /************** 判断并推送usdt的余额信息 ****************/
            if(array_key_exists('usdt',$list)) {
                $this->pushBalance($userId,$details['account_id'],$list['usdt']['trade']);
            }
            /************ END 判断并推送usdt的余额信息 **************/
            $tmpList = [];
            foreach($list as $key=>$value) {
                if($value['trade'] ==0 && $value['frozen'] == 0) {
                    continue;
                }
                $tmpList[] = implode('|',$value);
            }
            sort($tmpList);
            $listStr = implode(',',$tmpList);
            unset($tmpList);
            $validate = md5($listStr) . strlen($listStr);   // 获取验证字段

            if($accountDetails = $accouontTable->get($userId)) {
                // 判断数据未发生变化，直接返回成功
                if($accountDetails['validate'] == $validate && bccomp($accountDetails['amount_query_err_time'],0,3) != 1) {
                    return true;
                }
                if($accountDetails['validate'] != $validate) {
                    $changeData['balance'] = [];
                    // 存在上次获取的记录，本次进行对比
                    $changeData['balanceOld'] = $accountDetailsOld = self::explodeAccountDetails($accountDetails['details']);
                    $accountDetailsNew = self::explodeAccountDetails($listStr);
                    $codeList = array_unique(array_merge(array_keys($accountDetailsOld),array_keys($accountDetailsNew)));
                    foreach($codeList as $code) {
                        if(array_key_exists($code,$accountDetailsOld) && array_key_exists($code,$accountDetailsNew)) {
                            // 都存在，则为修改
                            if($accountDetailsOld[$code] == $accountDetailsNew[$code]) {
                                unset($changeData['balanceOld'][$code]);
                                continue;
                            }

                            // 获取修改的数据
                            $accountDetailsOld[$code] = explode('|',$accountDetailsOld[$code]);
                            $accountDetailsNew[$code] = explode('|',$accountDetailsNew[$code]);
                            $changeData['balance'][$code] = implode('|',[
                                CoinRuleRun::trimZero(bcsub($accountDetailsNew[$code][0],$accountDetailsOld[$code][0],self::SCALE)),
                                CoinRuleRun::trimZero(bcsub($accountDetailsNew[$code][1],$accountDetailsOld[$code][1],self::SCALE)),
                            ]);
                        } else if(array_key_exists($code,$accountDetailsOld)) { // 减少
                            $accountDetailsOld[$code] = explode('|',$accountDetailsOld[$code]);
                            foreach($accountDetailsOld[$code] as &$value) {
                                $value =  CoinRuleRun::trimZero(bcsub(0,$value,self::SCALE));
                            }
                            $changeData['balance'][$code] = implode('|',$accountDetailsOld[$code]);
                        } else {    // 增加
                            $changeData['balance'][$code] = $accountDetailsNew[$code];
                        }
                    }

                }
                if(bccomp($accountDetails['amount_query_err_time'],0,3) == 1) {
                    $changeData['queryError'] = false;
                }
                // 添加变更日志
                self::log('CHANGE',$userId,$changeData);
            } else {
                // 测试用
            //     $validate = '';
                $changeData = [
                    'balance'   => self::explodeAccountDetails($listStr)
                ];
                // 添加初始化日志
                self::log('INIT',$userId,$changeData);
            }

            // 更新本地保存记录情况
            $accountDetails = [
                'details'   => $listStr,
                'validate'  => $validate,
                'amount_query_err_time' => 0,
                'last_time' => microtime(true),
            ];
        } else {
            $changeData = [
                'queryError'    => true,
                'errorMsg'  => CallApi::getSaveError($balanceList),
            ];
            if($accountDetails = $accouontTable->get($userId)) {
                if(bccomp($accountDetails['amount_query_err_time'],0,3) == 1) {
                    // 原本就已是错误，无需修改
                    return true;
                }
                $accountDetails['amount_query_err_time'] = microtime(true);
                // 添加变更日志
                self::log('CHANGE',$userId,$changeData);
            } else {
                // 不存在上次记录，进行初始化添加
                $accountDetails = [
                    'details'   => '',
                    'validate'  => '',
                    'amount_query_err_time' => microtime(true),
                    'last_time' => microtime(true),
                ];
                // 添加初始化日志
                self::log('INIT',$userId,$changeData);
            }
        }
        if($accountDetails) {
            $accouontTable->set($userId,$accountDetails);
            // 通过 UDP 通知数据变更，执行修改处理
            self::noticeChange($userId,$changeData,preg_match('/usdt\|([0-9]+(\.[0-9]+)?)/',$accountDetails['details'],$matches)?$matches[1]:0);
        }
        return true;
    }

    /**
     * 分离用户账户详情
     * @param $details
     */
    static function explodeAccountDetails(string $details) {
        $result = [];
        $details = explode(',',$details);
        foreach ($details as $value) {
            $value = explode('|',$value);
            if(count($value) != 3) {
                continue;
            }
          //  $balance = rand(0,2) . '|' . rand(0,1);
            $balance = $value[1] . '|' . $value[2];
            if($balance == '0|0') {
                continue;
            }
            $result[$value[0]] = $balance;
        }
        return $result;
    }


    /**
     * 刷新用户记录
     * @param int $userId
     */
    private function userDetailsRefresh(int $userId = null) {
        /************* 获取表名列表 ***************/
        $tableNameAndCount = [
            'userDetails'    => 'USER_DETAILS',
            'processList'   => [],
        ];
        /******* 设置数量变化的表 **************/
        for($i = 0;$i < self::PROCESS_CONFIG['MONITOR_NUM'];++$i) {
            $tableName = self::PROCESS_CONFIG['TABLE_DYNAMIC_NAME'] . $i;
            $tableNameAndCount['processList'][$tableName] = self::swooleTable($tableName)->count();
        }
        /***** END 设置数量变化的表 ************/

        $monitorUserIdList = [];    //保存监控中的用户ID

        /******* 获取表名列表，并进行设置 *********/
        $where = [
            ['rule.is_open','=',CoinRule::STATUS_OPEN],
        ];
        if($userId) {
            $where[] = ['rule.user_id','=',$userId];
        }
        $maxRecord = self::PROCESS_CONFIG['SQL_ONCE_MAX_RECORD_NUM'];   //单次查询最大获取数
        $recordNum = $maxRecord;    //设置获取到的数量
        $i = 0; //循环次数
        while($recordNum == $maxRecord) {
            $list = Db::name('coin_rule')->alias('rule')
                ->join('users_key key','key.id=rule.user_id','INNER')
                ->field([
                    'key.user_id','key.account_id','key.access_key','key.secret_key',
                ])
                ->group('rule.user_id')
                ->limit($i++ * $maxRecord,$maxRecord)
                ->where($where)->select();
            if($recordNum = count($list)) {
                foreach($list as $userDetails) {
                    $this->userDetailsRefreshAddOne($tableNameAndCount,$userDetails);
                    $monitorUserIdList[] = $userDetails['user_id'];
                }
            }
        }
        Db::close();
        /***** END 获取需要实时更新的用户信息，并进行设置 *******/

        if(!is_null($userId) && $userId > 0) {
            // 单用户添加，不需要删除监控数据
            return true;
        }

        /*********** 删除不需要监控的用户数据 ************/
        $userTable = self::swooleTable($tableNameAndCount['userDetails']);
        $accountTable = self::swooleTable('ACCOUNTS_DETAILS');    //账户信息表
        foreach($userTable as $userId=>$userDetails) {
            if(in_array($userId,$monitorUserIdList)) {
                continue;   //跳出
            }
            // 删除监控数据
            $processTable = self::swooleTable($userDetails['process_index_name']);
            if($processTable->exists($userId)) {
                $processTable->del($userId);
            }
            if($accountTable->exists($userId)) {
                $accountTable->del($userId);
            }
            $userTable->del($userId);
            self::log('DEL',$userId);
        }
        /********* END 删除不需要监控的用户数据 **********/

        return true;
    }

    /**
     * 刷新添加用户数据信息
     * @param $tableNameAndCount 统计内容
     * @param $userDetails  用户详情
     */
    private function userDetailsRefreshAddOne(&$tableNameAndCount,$userDetails) {
        $isNewRecord = false;
        $userId = $userDetails['user_id'];
        $userTable = self::swooleTable($tableNameAndCount['userDetails']);
        /************* 获取processIndex ************/
        if($userTable->exists($userId) && ($user = $userTable->get($userId))) {
            $processIndexName = $user['process_index_name'];
        } else {
            $isNewRecord = true;
            $processList = $tableNameAndCount['processList'];
            asort($processList);
            foreach($processList as $processIndexName=>$num) {
                break;
            }
        }
        /*********** END 获取processIndex **********/
        $details = [
            'process_index_name'    => $processIndexName,
            'account_id'    => $userDetails['account_id'],
            'access_key'    => $userDetails['access_key'],
            'secret_key'    => $userDetails['secret_key'],
            'last_time'     => microtime(true),
        ];
        $userTable->set($userId,$details);
        self::swooleTable($details['process_index_name'])->set($userId,$details);
        if($isNewRecord) {
            $tableNameAndCount['processList'][$details['process_index_name']]++;
            self::log('ADD',$userId);
        }

        return true;
    }



    /**
     * 创建及获取表数据
     * @param null $tableKey
     * @param string|null $tableSuffix
     * @return bool|mixed
     */
    private static function swooleTable($tableKey = null,string $tableSuffix = null) {
        if(is_null($tableKey)) {    // 初始化表
            $tableList = self::getTableList();
            foreach($tableList as $table) {
                if(!$table[1]) {
                    continue;
                }
                // 创建Swoole表
                Swoole::table($table[0],$table[4],1024 * $table[3],$table[2]);
            }
            return true;
        }

        // 创建Swoole表
        $table = self::getTableList()[$tableKey];
        if($tableSuffix) {
            $table[0] .= $tableSuffix;
        }
        return Swoole::table($table[0],$table[4],1024 * $table[3],$table[2]);
    }



    /**
     * 获取表列表
     */
    static private function getTableList() {
        return array_merge(self::TABLE_LIST,self::$tableDynamicList);
    }


    /**
     * 日志记录
     * @param string $type 类型
     * @param $userId   用户ID
     * @param array $changeData 变更的数据
     */
    static private function log(string $type,$userId,array $changeData = null) {
        $changeData = is_array($changeData)?$changeData:[];
        $message = [
            self::PROCESS_CONFIG['LOG']['TYPE_CUSTOM_LIST'][$type],
            $userId,
            serialize($changeData),
        ];
        Log::write(implode(' ',$message),Log::LOG_TYPE_LIST[self::PROCESS_CONFIG['LOG']['TYPE_NAME']],false);
        return true;
    }

    /**
     * 通知信息变更
     * @param $userId 用户ID
     * @param $changeData   变更数据
     */
    static private function noticeChange($userId,$changeData,$usdt) {
        $params = [
            'userId'    => $userId,
            'apiStatus' => array_key_exists('queryError',$changeData) && $changeData['queryError']?"error":"ok",
            'usdt'  => null,
            'changeList'    => [],
        ];
        if($params['apiStatus'] == 'ok') {
            $params['usdt'] = $usdt;
            $params['changeList'] = array_key_exists('balance',$changeData)?array_keys($changeData['balance']):[];
        }
    //    print_r($params);
        // 发送请求处理
        UdpServer::request(self::$noticeConfig['CMD'],self::$noticeConfig['IP_LIST'],$params,self::$noticeConfig['TIMEOUT']);
    }

}